package com.wutoon.etl.handler;

import com.wutoon.etl.aop.PowerJobLogReport;
import com.wutoon.etl.constant.CommonConstant;
import com.wutoon.etl.mapper.BizOperateRecordManage;
import com.wutoon.etl.pojo.DataSourceDO;
import com.wutoon.etl.pojo.Db2HttpTask;
import com.wutoon.etl.pojo.Http2DbTask;
import com.wutoon.etl.pojo.MapReduceShard;
import com.wutoon.etl.util.DateUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicMarkableReference;
import java.util.stream.Collectors;

/**
 * @author 武兴云/72176468
 * @version 1.0
 * @date 2024/1/22 17:38
 * @description http采集存储db处理器
 */
@Setter
@Getter
@Slf4j
public class HttpPushHandler {
    private Db2HttpTask task;

    private RestTemplate restTemplate;

    private BizOperateRecordManage bizOperateRecordManage;

    /**
     * 阻塞队列
     */
    private BlockingQueue<Object[]> queue = new LinkedBlockingQueue<>(CommonConstant.MAX_FETCH_SIZE);

    private final ConcurrentHashMap<Integer, String> columnMap = new ConcurrentHashMap<>();

    private MapReduceShard shard;

    public HttpPushHandler(Db2HttpTask task, MapReduceShard shard, RestTemplate restTemplate, BizOperateRecordManage bizOperateRecordManage) {
        this.task = task;
        this.restTemplate = restTemplate;
        this.bizOperateRecordManage = bizOperateRecordManage;
        this.shard = shard;
    }
    /**
     * 参数示例：
     * {
     *      "shardNum": 1,
     *      "customParam": {
     *          "targetSource": {
     *              "url": "https://vwork-sit.vivo.xyz/universe-compute/data/collection/request",
     *              "method": "POST",
     *              "head": {
     *                  "Accept":"application/json"
     *              },
     *              "bodyTemplate":{
     *                  "code":"{code}",
     *                  "source":"vwork_sit_common_data_change:product",
     *                  "handleType":"INSERT",
     *                  "dataList":"{dataList}"
     *              },
     *              "data":{
     *                "key":"dataList",
     *                "position":"body/root,为root时直接推送数据列表，为body时推送内容为对象，列表作为对象的一个属性"
     *              },
     *              "requestNo":{
     *                "key":"code"
     *              },
     *              "judgeSuccessExpression":"code==200",
     *              "system":"PC_SERIALCODE_QUERY"
     *          },
     *          "sourceCode": "common_query",
     *          "querySql": "select id,tree_id,category,brand,subordinate_agency,source,sku_code,sku_name,sku_name_en,product_type,product_series,model_series,model_ver_id,marketing_id,product_abbr,model_name,material_serial,pre_sku,org_id,sort,agent_product_num,status,create_time,update_time,create_by,update_by,deleted_flag,lang_type,color_id,cgst,sgst,igst,hsn,pms_product_series,pms_product_model_series,sort_order from product",
     *          "batchInsertSize": 200,
     *          "concurrencySize": 1,
     *          "fetchSize":5000
     *      }
     * }
     *
     * @author 武兴云/72176468
     * @date 2024/2/29 15:43
     * @update_by 武兴云/72176468
     * @update_at 2024/2/29 15:43
     * @creed if you have doubt , please contact me !!!
     */
    public void start() throws Exception {
        PowerJobLogReport.report("HttpPushHandler start:{}", shard.getTaskName());
        try (Connection sourceConn = connect(task.getDataSource())) {
            // 部分参数做limit调整，例如：fetchSize不能超过2W
            limitParam(task);

            // 构建生产者
            ExecutorService executor = Executors.newFixedThreadPool(task.getConcurrencySize() + 1);
            CountDownLatch produceEnd = new CountDownLatch(CommonConstant.NUMBER_ONE);
            CountDownLatch consumerEnd = new CountDownLatch(task.getConcurrencySize());
            // 主线程和子线程之间消息传递
            AtomicMarkableReference<String> msg = new AtomicMarkableReference<>(null, false);
            JdbcReader reader = new JdbcReader(queue, columnMap, task, sourceConn, produceEnd, msg, RequestContextHolder.getRequestAttributes());
            executor.execute(reader);

            RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
            // 4、构建消费者
            for(int i = 0; i < task.getConcurrencySize(); i ++) {
                HttpPusher pusher = new HttpPusher(restTemplate, requestAttributes, produceEnd, queue, columnMap, task, consumerEnd, msg, bizOperateRecordManage);
                executor.execute((pusher));
            }
            // 6、关闭连接
            long start = System.currentTimeMillis();
            long now = System.currentTimeMillis();
            // 循环观测子线程是否存在异常，及时感知，否则发生异常时感知不及时，会等到全部结束或一小时才结束任务
            while ((now - start) / 1000 < CommonConstant.CONSUMER_TIME_OUT) {
                // 消费结束或存在异常，退出，否则继续观测
                if (consumerEnd.await(CommonConstant.LOOP_OBSERVATION_TIME_OUT, TimeUnit.SECONDS) || msg.isMarked()) {
                    break;
                }
                now = System.currentTimeMillis();
            }
            if (!msg.isMarked()) {
                log.info("DataTransferHandler end success");
                // 正常关闭线程池
                executor.shutdown();
            } else {
                log.warn("DataTransferHandler not end in {} seconds, shutdown now", CommonConstant.CONSUMER_TIME_OUT);
                // 关闭当前正在执行的子线程，只会interrupt正在执行的线程，需要在子线程判断终止标志
                executor.shutdownNow();
                throw new RuntimeException(msg.getReference());
            }
        } catch (Exception e) {
            log.error("DataTransferHandler error", e);
            throw e;
        }
    }

    /**
     * 目标预执行sql，主要是清空源表
     *
     * @param targetConnections List<Connection>
     * @param task Http2DbTask
     * @author 武兴云/72176468
     * @date 2024/2/27 15:24
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:24
     * @creed if you have doubt , please contact me !!!
     */
    private void executePreSQl(List<Connection> targetConnections, Http2DbTask task) {
        // 全量场景需要清空目标表
        if (!task.getIncrease()) {
            log.info("DataTransferHandler execute pre truncate sql");

            // 重命名旧表
            String tableBackUp = task.getTargetTable() + DateUtil.generateCurrentDayId(DateUtil.MMDDHHMM);
            String renameTableSql = getRenameTableSql(task, tableBackUp);
            try (PreparedStatement stmt = targetConnections.get(0).prepareStatement(renameTableSql)) {
                stmt.executeUpdate();
            } catch (Exception e) {
                log.error("truncate table error:{}", renameTableSql, e);
                throw new RuntimeException("truncate table error:" + renameTableSql + e.getMessage());
            }

            // 根据目标表创建新表，
            String createNewTableSql = getCreateNewTableSql(task, tableBackUp);
            try (PreparedStatement stmt = targetConnections.get(0).prepareStatement(createNewTableSql)) {
                stmt.executeUpdate();
            } catch (Exception e) {
                log.error("truncate table error:{}", createNewTableSql, e);
                throw new RuntimeException("truncate table error:" + createNewTableSql + e.getMessage());
            }
        }
    }

    /**
     * 获取重命名表sql
     *
     * @param task Http2DbTask
     * @param tableBackUp String
     * @return String
     * @author 武兴云/72176468
     * @date 2024/2/27 15:24
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:24
     * @creed if you have doubt , please contact me !!!
     */
    private static String getRenameTableSql(Http2DbTask task, String tableBackUp) {
        String renameTableSql;
        switch (task.getTargetSource().getDriver()) {
            // 默认是mysql语法
            case CommonConstant.MYSQL_DRIVER:
            default:
                renameTableSql = String.format(CommonConstant.RENAME_TABLE, task.getTargetTable(), tableBackUp);
        }
        return renameTableSql;
    }
    /**
     * 获取创建表sql
     *
     * @param task Http2DbTask
     * @param tableBackUp String
     * @return String
     * @author 武兴云/72176468
     * @date 2024/2/27 15:23
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:23
     * @creed if you have doubt , please contact me !!!
     */
    private static String getCreateNewTableSql(Http2DbTask task, String tableBackUp) {
        String renameTableSql;
        switch (task.getTargetSource().getDriver()) {
            // 默认是mysql语法
            case CommonConstant.MYSQL_DRIVER:
            default:
                renameTableSql = String.format(CommonConstant.CREATE_TABLE_LIKE, task.getTargetTable(), tableBackUp);
        }
        return renameTableSql;
    }

    /**
     * 关闭连接
     *
     * @param targetConnections List<Connection>
     * @author 武兴云/72176468
     * @date 2024/2/27 15:23
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:23
     * @creed if you have doubt , please contact me !!!
     */
    private void closeConnections(List<Connection> targetConnections) {
        for (Connection connection : targetConnections) {
            try {
                connection.close();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 参数限制重写
     *
     * @param task Http2DbTask
     * @author 武兴云/72176468
     * @date 2024/2/27 15:23
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:23
     * @creed if you have doubt , please contact me !!!
     */
    private void limitParam(Db2HttpTask task) {
        // 部分值做限制
        Db2HttpTask.HttpPushSourceConfig targetSource = task.getTargetSourceConfig();
        // 采集最大并发限制
        if (task.getConcurrencySize() == null || task.getConcurrencySize() > CommonConstant.MAX_CONCURRENCY_SIZE
                || task.getConcurrencySize() < CommonConstant.NUMBER_ZERO) {
            task.setConcurrencySize(CommonConstant.DEFAULT_PUSH_CONCURRENCY_SIZE);
        }
        // 最大写入批次大小
        if (task.getBatchInsertSize() > CommonConstant.MAX_BATCH_SIZE || task.getBatchInsertSize() < CommonConstant.NUMBER_ZERO) {
            task.setBatchInsertSize(CommonConstant.MAX_BATCH_SIZE);
        }
    }

    /**
     * 获取链接
     *
     * @param dataSourceDO DataSourceDO
     * @return Connection
     * @author 武兴云/72176468
     * @date 2024/2/27 15:23
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 15:23
     * @creed if you have doubt , please contact me !!!
     */
    private Connection connect(DataSourceDO dataSourceDO) {
        try {
            Class.forName(dataSourceDO.getDriver());
            DriverManager.setLoginTimeout(CommonConstant.TIMEOUT_SECONDS);
            return DriverManager.getConnection(dataSourceDO.getJdbcUrl(), dataSourceDO.getUsername(), dataSourceDO.getPassword());
        } catch (Exception e) {
            throw new RuntimeException("initialize datasource connection error");
        }
    }

    /**
     * 获取查询维度列表
     *
     * @param sourceConfig Http2DbTHttpCollectSourceConfig
     * @return List<String>
     * @author 武兴云/72176468
     * @date 2024/2/27 11:53
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 11:53
     * @creed if you have doubt , please contact me !!!
     */
    private List<String> getDimensionSet(Http2DbTask.HttpCollectSourceConfig sourceConfig) {
        String set = sourceConfig.getDimension().getValue();
        if (set.startsWith(CommonConstant.REF_METHOD)) {
            // 获取方法对象
            Method method;
            try {
                method = this.getClass().getMethod(set.split(CommonConstant.REF_METHOD)[1].trim());
                return (List<String>) method.invoke(this);
            } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
                log.error("Http collector getDimensionSet error");
                throw new RuntimeException(e);
            }
        }
        return Arrays.stream(set.split(",")).map(String::trim).collect(Collectors.toList());
    }

    /**
     * 测试方法
     *
     * @return List<String>
     * @author 武兴云/72176468
     * @date 2024/2/27 11:52
     * @update_by 武兴云/72176468
     * @update_at 2024/2/27 11:52
     * @creed if you have doubt , please contact me !!!
     */
    public List<String> testGetDimensionSet() {
        return Collections.singletonList("3");
    }

    static class CustomBlockingPolicy extends ThreadPoolExecutor.AbortPolicy {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            if (!executor.isShutdown()) {
                try {
                    // 阻塞等待队列有空闲位置
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
